package kcp;

import com.backblaze.erasure.fec.Fec;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollChannelOption;
import io.netty.channel.epoll.EpollDatagramChannel;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.kqueue.KQueue;
import io.netty.channel.kqueue.KQueueDatagramChannel;
import io.netty.channel.kqueue.KQueueEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.util.HashedWheelTimer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.Getter;
import threadPool.IMessageExecutorPool;

/** Created by JinMiao 2018/9/20. */
public class KcpServer {

  private IMessageExecutorPool iMessageExecutorPool;

  private EventLoopGroup group;

  private final List<Channel> localAddressList = new ArrayList<>();

  private HashedWheelTimer hashedWheelTimer;

  @Getter private IChannelManager channelManager;

  public void init(KcpListener kcpListener, ChannelConfig channelConfig, int... ports) {
    if (channelConfig.isUseConvChannel()) {
      int convIndex = 0;
      if (channelConfig.getFecAdapt() != null) {
        convIndex += Fec.fecHeaderSizePlus2;
      }
      channelManager = new ServerConvChannelManager(convIndex);
    } else {
      channelManager = new ServerAddressChannelManager();
    }

    hashedWheelTimer = new HashedWheelTimer(new TimerThreadFactory(), 1, TimeUnit.MILLISECONDS);

    this.iMessageExecutorPool = channelConfig.getIMessageExecutorPool();

    int cpuNum = Runtime.getRuntime().availableProcessors();
    int bindTimes = 1;

    boolean epoll = Epoll.isAvailable();
    boolean kqueue = KQueue.isAvailable();

    Bootstrap bootstrap = new Bootstrap();
    if (epoll || kqueue) {
      // ADD SO_REUSEPORT ？ https://www.jianshu.com/p/61df929aa98b
      bootstrap.option(EpollChannelOption.SO_REUSEPORT, true);
      bindTimes = cpuNum;
    }
    Class<? extends Channel> channelClass = null;
    if (epoll) {
      group = new EpollEventLoopGroup(cpuNum);
      channelClass = EpollDatagramChannel.class;
    } else if (kqueue) {
      group = new KQueueEventLoopGroup(cpuNum);
      channelClass = KQueueDatagramChannel.class;
    } else {
      group = new NioEventLoopGroup(ports.length);
      channelClass = NioDatagramChannel.class;
    }

    bootstrap.channel(channelClass);
    bootstrap.group(group);
    bootstrap.handler(
        new ChannelInitializer<Channel>() {
          @Override
          protected void initChannel(Channel ch) {
            ServerChannelHandler serverChannelHandler =
                new ServerChannelHandler(
                    channelManager,
                    channelConfig,
                    iMessageExecutorPool,
                    kcpListener,
                    hashedWheelTimer);
            ChannelPipeline cp = ch.pipeline();
            if (channelConfig.isCrc32Check()) {
              Crc32Encoder crc32Encoder = new Crc32Encoder();
              Crc32Decoder crc32Decoder = new Crc32Decoder();
              // 这里的crc32放在eventloop网络线程处理的，以后内核有丢包可以优化到单独的一个线程处理
              cp.addLast(crc32Encoder);
              cp.addLast(crc32Decoder);
            }
            cp.addLast(serverChannelHandler);
          }
        });
    // bootstrap.option(ChannelOption.SO_RCVBUF, 10*1024*1024);
    bootstrap.option(ChannelOption.SO_REUSEADDR, true);

    for (int port : ports) {
      for (int i = 0; i < bindTimes; i++) {
        ChannelFuture channelFuture = bootstrap.bind(port);
        Channel channel = channelFuture.channel();
        localAddressList.add(channel);
      }
    }

    Runtime.getRuntime().addShutdownHook(new Thread(this::stop));
  }

  // public void init(int workSize, KcpListener kcpListener, ChannelConfig channelConfig, int...
  // ports) {
  //    DisruptorExecutorPool disruptorExecutorPool = new DisruptorExecutorPool();
  //    for (int i = 0; i < workSize; i++) {
  //        disruptorExecutorPool.createDisruptorProcessor("disruptorExecutorPool" + i);
  //    }
  //    init(disruptorExecutorPool, kcpListener, channelConfig, ports);
  // }

  public void stop() {
    localAddressList.forEach(ChannelOutboundInvoker::close);
    channelManager.getAll().forEach(Ukcp::close);
    if (iMessageExecutorPool != null) {
      iMessageExecutorPool.stop();
    }
    if (hashedWheelTimer != null) {
      hashedWheelTimer.stop();
    }
    if (group != null) {
      group.shutdownGracefully();
    }
  }

  /** 定时器线程工厂* */
  private static class TimerThreadFactory implements ThreadFactory {
    private final AtomicInteger timeThreadCounter = new AtomicInteger(0);

    @Override
    public Thread newThread(Runnable r) {
      return new Thread(r, "KcpServerTimerThread " + timeThreadCounter.addAndGet(1));
    }
  }
}
